
package org.zjt.spark.mllib

import org.apache.spark.{SparkConf, SparkContext}
// $example on$
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating
// $example off$

object RecommendationExample {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("CollaborativeFilteringExample").setMaster("local[2]")
    val sc = new SparkContext(conf)
    // $example on$
    // Load and parse the data    1,1,5.0
    val data = sc.textFile("/Users/zhangjuntao/IdeaProjects/myproject/hw-bigdata/scala-demo/src/main/resource/mllib/als/test.data")
    val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
      Rating(user.toInt, item.toInt, rate.toDouble)
    })

    // Build the recommendation model using ALS
    val rank = 10
    val numIterations = 10
    val model = ALS.train(ratings, rank, numIterations, 0.01)

    // Evaluate the model on rating data  删除该用户喜爱的rate程度
    val usersProducts = ratings.map { case Rating(user, product, rate) =>
      (user, product)
    }

    //得到model的喜爱度的预测
    val predictions = model.predict(usersProducts).map { case Rating(user, product, rate) =>
        ((user, product), rate)
      }


    //将得到的喜爱预期和原始的比较
    val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
      ((user, product), rate)
    }.join(predictions).persist()


    println(ratesAndPreds.collect().mkString("\n"))

    // 均方误差（Mean Squared Error, MSE）
    val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
      val err = (r1 - r2)
      err * err
    }.mean()
    println("Mean Squared Error = " + MSE)

    // 保存、加载模型
    model.save(sc, "target/tmp/myCollaborativeFilter")
    val sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")

    //释放内存
    ratings.unpersist(blocking = false)


    // 得到用户1 的前两个推荐
    println(sameModel.recommendProducts(1,2).mkString(","))


    try {
      Thread.sleep(200000)
    }catch {
      case e: Exception => e.printStackTrace()
    }finally {
      sc.stop()
    }

  }
}
